package com.alibaba.otter.node.etl.load.loader.mq.rabbit;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.api.ChannelManager;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.RabbitMqConfirmListener;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.callback.SuccessCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;

@Slf4j
public class ProxiedConfirmListener implements ConfirmListener {

    @Setter
    private RabbitMqConfirmListener target = new RabbitMqConfirmListener();

    @Setter
    private Channel channel;

    @Setter
    private String namespace;

    @Setter
    private ChannelManager channelManager;

    @Override
    public void handleAck(long deliveryTag, boolean multiple) {
        try {
            target.handleAck(deliveryTag, multiple);
        } catch (Exception e) {
            log.error("Failed to handle ack. channelNumber: " +
                channel.getChannelNumber() + ", deliveryTag: " + deliveryTag +
                ", toConfirms: " + target.getConfirmMap(), e);
        } finally {
            channelManager.releaseChannel(channel, namespace);
        }
    }

    @Override
    public void handleNack(long deliveryTag, boolean multiple) {
        try {
            target.handleNack(deliveryTag, multiple);
        } catch (Exception e) {
            log.error("Failed to handle nack. channelNumber: " +
                channel.getChannelNumber() + ", deliveryTag: " + deliveryTag +
                ", toConfirms: " + target.getConfirmMap(), e);
        } finally {
            channelManager.releaseChannel(channel, namespace);
            log.error("unconfirmed, deliveryTag: " + deliveryTag + ", channelNumber: " + channel.getChannelNumber());
        }
    }

    public void toConfirm(long deliveryTag, Object message) {
        target.toConfirm(deliveryTag, message);
    }

    public void remove(long deliveryTag) {
        target.remove(deliveryTag);
    }

    public void addSuccessCallbacks(List<SuccessCallback> successCallbacks) {
        target.addSuccessCallback(successCallbacks);
    }
}
